/*
 *
 *  * Copyright 2015-2101 Shouqi Limousine & chauffeur Group.
 *  *
 *  * Licensed under the Apache License, Version 2.0 (the "License");
 *  * you may not use this file except in compliance with the License.
 *  * You may obtain a copy of the License at
 *  *
 *  *      http://www.apache.org/licenses/LICENSE-2.0
 *  *
 *  * Unless required by applicable law or agreed to in writing, software
 *  * distributed under the License is distributed on an "AS IS" BASIS,
 *  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  * See the License for the specific language governing permissions and
 *  * limitations under the License.
 *
 */

package com.sq.plugin.redis.client;

import com.fasterxml.jackson.core.type.TypeReference;
import com.sq.plugin.redis.serializer.exception.SerializationException;
import com.sq.plugin.redis.serializer.impl.jackson.Jackson2JsonSerializer;
import com.sq.plugin.redis.serializer.util.ByteUtils;
import com.sq.plugin.redis.serializer.util.SerializationUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisSentinelPool;

import java.util.*;

/**
 * @author wangliang
 * @date 2018/5/15
 */
public class RedisSentinelClient extends BaseRedisClient {
    private Logger logger = LoggerFactory.getLogger(RedisSentinelClient.class);

    /**
     * 哨兵redis连接池
     */
    private JedisSentinelPool sentinelJedisPool;

    /**
     * 连接池对应map
     */
    private List<JedisSentinelPool> sentinelJedisPools;

    /**
     * 默认构造方法
     */
    public RedisSentinelClient(JedisSentinelPool defaultPool,List<JedisSentinelPool> sentinelJedisPools) {

        this.sentinelJedisPool = defaultPool;
        this.sentinelJedisPools = sentinelJedisPools;
        serializerInit(DEFAULT_SERIALIZER);
    }

    private int hash(String key) {
        try {
            int feed = _hash(key);
            int nodeCount = this.sentinelJedisPools.size();
            return feed % nodeCount + 1;
        } catch (Exception e) {
            logger.error("hash key:{} throws Exception;{}", key, e);
            return 1;
        }
    }

    /**
     * hashMap的hash算法
     */
    private int _hash(Object key) {
        int h;
        return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);
    }


    private JedisSentinelPool getJedisSentinelPool(String key) {
        JedisSentinelPool pool;
        try {
            int id = hash(key);
            pool = this.sentinelJedisPools.get(id);
            if (pool == null) {
                pool = this.sentinelJedisPool;
            }
        } catch (Exception e) {
            pool = this.sentinelJedisPool;
        }
        return pool;
    }

    private Jedis getShardedJedis(String key) {
        JedisSentinelPool pool;
        try {
            pool = getJedisSentinelPool(key);
        } catch (Exception e) {
            return this.sentinelJedisPool.getResource();
        }
        if (pool != null) {
            return pool.getResource();
        }
        return this.sentinelJedisPool.getResource();
    }

    private void returnResource(Jedis shardedJedis) {
        shardedJedis.close();
    }

    @Override
    public String set(String key, Object value) {
        Jedis jedis = getShardedJedis(key);
        try {
            byte[] byteArray = ByteUtils.objectToByte(value);
            return jedis.set(key.getBytes(), byteArray);
        } catch (Exception e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }


    @Override
    public String set(String key, int expTime, Object value) {
        Jedis jedis = getShardedJedis(key);
        String result = null;
        try {
            result = jedis.setex(key.getBytes(), expTime, ByteUtils.objectToByte(value));
        } catch (Exception e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return result;
    }

    @Override
    public String set(String key, int expTime, String value) {
        Jedis jedis = getShardedJedis(key);
        String result = null;
        try {
            result = jedis.setex(getKey(key), expTime, serializer.serialize(value));
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return result;
    }

    @Override
    public <T> T getSet(final String key, final Object value, Class<T> javaType) {
        Jedis jedis = getShardedJedis(key);
        byte[] data = jedis.getSet(getKey(key), serializer.serialize(value));
        try {
            return serializer.deserialize(data, javaType);
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    @Override
    public boolean set(final String key, final Object value, final String nxxx, final String expx, final long time) {
        Jedis jedis = getShardedJedis(key);
        try {
            return jedis.set(getKey(key), serializer.serialize(value), nxxx.getBytes(), expx.getBytes(), time) != null;
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return false;
    }

    @Override
    public <T> T get(final String key, Class<T> javaType) {
        Jedis jedis = getShardedJedis(key);
        try {
            return serializer.deserialize(jedis.get(getKey(key)), javaType);
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    /**
     * 当serializerType不等于DEFAULT_SERIALIZER时,
     * 使用该方法会抛出异常
     */
    @Override
    public <T> T get(final String key, TypeReference<T> tr) {
        Jedis jedis = getShardedJedis(key);
        try {
            byte[] value = jedis.get(getKey(key));
            if (serializer instanceof Jackson2JsonSerializer) {
                return ((Jackson2JsonSerializer) serializer).deserialize(value, tr);
            }
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    @Override
    public String get(final String key) {
        Jedis jedis = getShardedJedis(key);
        try {
            return serializer.deserialize(jedis.get(getKey(key)), String.class);
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    @Override
    public long delete(final String key) {
        Jedis jedis = getShardedJedis(key);
        try {
            return jedis.del(getKey(key));
        } catch (Exception e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return -1;
    }

    @Override
    public byte[] getKey(String key) {
        return SerializationUtils.encode(key);
    }

    /**
     * getKey的逆操作 获取原始key值
     */
    private String getOriginalKey(byte[] key) {
        return new String(key);
    }

    @Override
    public byte[] hget(final String key, final String field) {
        Jedis jedis = getShardedJedis(key);

        try {
            return jedis.hget(getKey(key), getKey(field));
        } catch (Exception e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    @Override
    public long hincrBy(final String key, final String field, final long value) {
        Jedis jedis = getShardedJedis(key);

        try {
            return jedis.hincrBy(getKey(key), getKey(field), value);
        } catch (Exception e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return -1;
    }

    @Override
    public long incrBy(final String key, final long delta) {
        Jedis jedis = getShardedJedis(key);

        try {
            return jedis.incrBy(getKey(key), delta);
        } catch (Exception e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return -1;
    }

    @Override
    public long expire(final String key, final int seconds) {
        Jedis jedis = getShardedJedis(key);

        try {
            return jedis.expire(getKey(key), seconds);
        } catch (Exception e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return -1;
    }

    @Override
    public long rpush(final String key, final Object value) {
        Jedis jedis = getShardedJedis(key);

        try {
            return jedis.rpush(getKey(key), serializer.serialize(value));
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return -1;
    }

    @Override
    public <T> T lpop(final String key, Class<T> javaType) {
        Jedis jedis = getShardedJedis(key);

        try {
            return serializer.deserialize(jedis.lpop(getKey(key)), javaType);
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    @Override
    public <T> T blpop(final String key, final int timeout, Class<T> javaType) {
        Jedis jedis = getShardedJedis(key);

        try {
            List<byte[]> list = jedis.blpop(timeout, getKey(key));
            if (list == null || list.isEmpty()) {
                return null;
            } else {
                return serializer.deserialize(list.get(1), javaType);
            }
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    @Override
    public <T> List<T> lrange(final String key, int start, int end) {
        Jedis jedis = getShardedJedis(key);

        try {
            List<byte[]> list = jedis.lrange(getKey(key), start, end);

            return (List) deserializeValues(list, List.class);
        } catch (Exception e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    @Override
    public Long sadd4Sets(final String key, final Object... values) {
        Jedis jedis = getShardedJedis(key);

        if (values == null) {
            return -1L;
        }

        try {
            byte[][] valBytes = new byte[values.length][];
            for (int i = 0; i < values.length; i++) {
                valBytes[i] = serializer.serialize(values[i]);
            }

            return jedis.sadd(getKey(key), valBytes);
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    @Override
    public <T> Set<T> smembers4Sets(final String key, Class<T> javaType) {
        Jedis jedis = getShardedJedis(key);

        try {
            Set<byte[]> resultSet = jedis.smembers(getKey(key));
            Set<T> result = new HashSet<T>();
            if (resultSet == null) {
                return result;
            } else {
                for (byte[] b : resultSet) {
                    result.add(serializer.deserialize(b, javaType));
                }
            }

            return result;
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    @Override
    public Long srem4Sets(final String key, final Object... values) {
        if (values == null) {
            return -1L;
        }
        Jedis jedis = getShardedJedis(key);

        try {
            byte[][] valBytes = new byte[values.length][];
            for (int i = 0; i < values.length; i++) {
                valBytes[i] = serializer.serialize(values[i]);
            }

            return jedis.srem(getKey(key), valBytes);
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return -1L;
    }

    @Override
    public boolean exists(String key) {
        Jedis jedis = getShardedJedis(key);

        try {
            return jedis.exists(getKey(key));
        } catch (Exception e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return false;
    }

    @Override
    public long hdel(String key, String... field) {
        Jedis jedis = getShardedJedis(key);

        try {
            if (field.length > 0) {
                byte[][] fieldArray = new byte[field.length][];
                for (int i = 0; i < field.length; i++) {
                    fieldArray[i] = getKey(field[i]);
                }
                return jedis.hdel(getKey(key), fieldArray);
            } else {
                throw new SerializationException("ERR wrong number of arguments for 'hdel' command");
            }
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return -1;
    }

    /**
     * 返回哈希表 key 中，所有的域和值
     */
    @Override
    public Map<String, Object> hgetAll(String key) {
        Jedis jedis = getShardedJedis(key);

        try {
            Map<byte[], byte[]> map = jedis.hgetAll(getKey(key));
            Map<String, Object> res = new HashMap<>(map.size());
            Set<Map.Entry<byte[], byte[]>> entries = map.entrySet();
            for (Map.Entry<byte[], byte[]> e : entries) {
                res.put(getOriginalKey(e.getKey()), serializer.deserialize(e.getValue(), Object.class));
            }
            return res;
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    /**
     * 同时将多个 field-value (域-值)对设置到哈希表 key 中
     */
    @Override
    public String hmset(String key, Map<String, Object> map) {

        Jedis jedis = getShardedJedis(key);

        try {
            return jedis.hmset(getKey(key), serializerMap(map));
        } catch (Exception e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    /**
     * 返回哈希表 key 中，一个或多个给定域的值
     */
    @Override
    public List<Object> hmget(String key, String... field) {
        Jedis jedis = getShardedJedis(key);
        try {
            if (field.length > 0) {
                byte[][] fieldArray = new byte[field.length][];
                for (int i = 0; i < field.length; i++) {
                    fieldArray[i] = SerializationUtils.encode(field[i]);
                }
                List<Object> res = new ArrayList<>(field.length);
                List<byte[]> values = jedis.hmget(getKey(key), fieldArray);
                for (byte[] v : values) {
                    res.add(serializer.deserialize(v, Object.class));
                }
                return res;
            } else {
                throw new SerializationException("ERR wrong number of arguments for 'hmget' command");
            }
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    /**
     * 返回哈希表 key 中所有域的值
     */
    @Override
    public <T> List<T> hvals(String key) {
        Jedis jedis = getShardedJedis(key);

        try {
            Collection<byte[]> hvals = jedis.hvals(getKey(key));
            return (List) deserializeValues(hvals, List.class);
        } catch (Exception e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    /**
     * 返回哈希表 key 中的所有域
     */
    @Override
    public Set<String> hkeys(String key) {
        Jedis jedis = getShardedJedis(key);

        try {
            Set<byte[]> hkeys = jedis.hkeys(getKey(key));
            Set<String> res = new HashSet<>(hkeys.size());
            for (byte[] k : hkeys) {
                res.add(getOriginalKey(k));
            }
            return res;
        } catch (Exception e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    @Override
    public Long setnx(String key, Object value) {
        Jedis jedis = getShardedJedis(key);
        try {
            return jedis.setnx(getKey(key), serializer.serialize(value));
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return -1L;
    }

    @Override
    public Long llen(String key) {
        Jedis jedis = getShardedJedis(key);

        try {
            return jedis.llen(getKey(key));
        } catch (Exception e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return -1L;
    }

    @Override
    public Long lrem(String key, long count, Object value) {
        Jedis jedis = getShardedJedis(key);

        try {
            return jedis.lrem(getKey(key), count, serializer.serialize(value));
        } catch (SerializationException e) {
            logger.error("ERROR!\n{}", e);
        } finally {
            returnResource(jedis);
        }
        return -1L;
    }
}
